package com.zuikaku.demo;

import com.zuikaku.pojo.Order;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.math.BigDecimal;
import java.util.Date;
import java.util.UUID;

public class _2KeyByFilterReduceDemo {
    public static void main(String[] args) {
        //创建环境
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.createLocalEnvironment();
        environment.setParallelism(1);
        //根据order获取source
        DataStreamSource<Order> orderDS = environment.fromElements(new Order(UUID.randomUUID().toString(),"HuaWei Mate60",1001,new BigDecimal("1000"),new Date()),
                new Order(UUID.randomUUID().toString(),"iPhone 15 Pro Max",1002,new BigDecimal("800"),new Date()),
                new Order(UUID.randomUUID().toString(),"HuaWei Mate60",1001,new BigDecimal("990"),new Date()),
                new Order(UUID.randomUUID().toString(),"Nokia X1",1002,new BigDecimal("700"),new Date()),
                new Order(UUID.randomUUID().toString(),"Nokia X1",1003,new BigDecimal("700"),new Date()));


        //根据itemName和customerId进行分组
        KeyedStream<Order, Tuple2<String, Integer>> keyDs = orderDS.keyBy(new KeySelector<Order, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> getKey(Order order) throws Exception {
                return Tuple2.of(order.getItemName(),order.getCustomerId());
            }
        });

        //然后金额求和
        SingleOutputStreamOperator<Order> reduce = keyDs.reduce(new ReduceFunction<Order>() {
            @Override
            public Order reduce(Order left, Order right) throws Exception {
                right.setPrice(right.getPrice().add(left.getPrice()));
                return right;
            }
        });

        //filter不看用户1003的
        SingleOutputStreamOperator<Order> filterDs = reduce.filter(new FilterFunction<Order>() {
            @Override
            public boolean filter(Order order) throws Exception {
                return order.getCustomerId()!=1003;
            }
        });
        filterDs.print("处理后");
        //4.开启job(job默认并行执行，且无顺序)
        try {
            environment.execute("myJob");
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
